const Queue = require('bull');
const redis = require('ioredis');
const queueMap = new Map;
const subMap = new Map;
let queueConfig = {};
let subscriber = null;
let publisher = null;



module.exports.process = function (name, func) {
    let { queue } = getQueue(name);
    return queue.process(func);
}


module.exports.getJob = function (name) {
    let { queue } = getQueue(name);
    return queue.getNextJob();
}

module.exports.get = function (name) {
    return publisher.get(name)
}

module.exports.set = function (name,v) {
    return publisher.set(name,v)
}

module.exports.getJob = function (name) {
    let { queue } = getQueue(name);
    return queue.getNextJob();
}

module.exports.init = function ({ host = "", port = "", password = "", db=0 }) {
    queueConfig.host = host;
    queueConfig.port = port;
    queueConfig.password = password;
    queueConfig.db = db;
    console.log('queueConfig..', queueConfig)
    publisher = redis.createClient(queueConfig);
    subscriber = redis.createClient(queueConfig);
    let num = 0;
    let self = this;

    return new Promise((res, rej) => {
        publisher.on("connect", async function () {
            self.client = publisher;
            if (++num>=2) {
                res(self)
            }

        })
        subscriber.on("connect", function () {
            num++;
            subscriber.on("message", function (channel, message) {
                let funcs = subMap.get(channel);
                if (Array.isArray(funcs)) {
                    funcs.forEach(f => {
                        f(message)
                    })
                }
            });
            if (++num>=2) {
                res(self)
            }
        })
    })

}

//订阅
module.exports.subscribe = function (channel, func) {

    if (typeof func !== 'function') throw "参数必须为函数";
    if (subMap.has(channel)) {
        listenerList = subMap.get(channel).push(func)
    } else {
        subMap.set(channel, listenerList = [func])
    }
    subscriber.subscribe(channel);
}


//发布
module.exports.publish = function (channel, data) {
    publisher.publish(channel, data);
}


//订阅一个消息
module.exports.on = function (name, func) {

    let { queue, listenerMap } = getQueue(name);


    let funcs = {
        async completed(jobId, result) {
            
            let job = await queue.getJob(jobId);

            result = JSON.parse(result);

            let info =  null;

            if(typeof job.data.info ==='string'){
                info = JSON.parse(job.data.info);
            }

            if (typeof result === 'object' && result.status === "discard") {
                return
            }


            func({
                id: job.id,
                status: 'done',
                result: result,
                error: null,
                info
            })
        },
        failed(job, error) {


            let info =  null;

            if(typeof job.data.info ==='string'){
                info = JSON.parse(job.data.info);
            }
            func({
                id: job.id,
                status: 'fail',
                result: null,
                error,
                info
            })
        },
        error(e) {
            func({
                id: " job.id",
                status: 'error',
                result: null,
                e,
                info: e
            })
        }
    };

    listenerMap.set(func, funcs);

    ['completed', 'failed', 'error'].forEach(k => {
        queue.on("global:" + k, funcs[k])
    });


}

//取消订阅
module.exports.off = function (name, func) {

    let { queue, listenerMap } = getQueue(name);
    let funcs = listenerMap.get(func);

    if (!funcs) return;

    ['completed', 'failed'].forEach(k => {
        queue.off(k, funcs[k])
    });
    funcs.delete(func);
}


//触发一个消息, info 附加信息原样返回到订阅。
module.exports.emit = function (name, message, info, callbackUrl) {

    let { queue } = getQueue(name);
    return queue.add({ message, info })
}


function getQueue(name) {
    if (queueMap.has(name)) {
        return queueMap.get(name);
    } else {
        let queue = new Queue(name, { redis: queueConfig });
        let rel;
        queueMap.set(name, rel = { listenerMap: new Map, queue });

        return rel
    }
}

